package Lothar

import org.apache.spark.sql.functions.col
import org.apache.spark.sql.{DataFrame, SparkSession}

/*
* 从Step1中大概知道数据的情况后
* Step2进行数据预处理，给后面的SVM(支持向量机)训练
* */

object Step2 {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .config("hive.metastore.uris", "thrift://master:9083")
      .config("spark.sql.warehouse.dir", "hdfs://master:8020/usr/hive/warehouse")
      .enableHiveSupport()
      .getOrCreate()

    var MediaIndexDF= spark.read.table("portrait.media_index")
    val BilleventDF = spark.read.table("portrait.billevent")
    val OrderDF = spark.read.table("portrait.orderData")
    val UserMsgDF = spark.read.table("portrait.usermsg")
    val UserEventDF = spark.read.table("portrait.userevent")

    // where进入是面向DataSet的，以前真没注意。
    val MediaIndex: DataFrame = filterOCode(filterOName(MediaIndexDF))
      .where("duration>4000 & duration<21600000")
      .filter("res_type !=0 or origin_time not like '00%' or origin_time not like '00%'")

    val Billevent: DataFrame = filterOCode(filterOName(BilleventDF))
    val Order: DataFrame = filterOCode(filterOName(OrderDF))
    val UserMsg: DataFrame = filterOCode(filterOName(UserMsgDF)).filter(col("")==="正常" || col("")==="欠费暂停"
      || col("")==="主动暂停"||col("")==="主动销户")
    val UserEvent: DataFrame = filterOCode(filterOName(UserEventDF))

    MediaIndex.write.mode("overwrite").saveAsTable("portrait.MediaIndex1")
    Billevent.write.mode("overwrite").saveAsTable("portrait.Billevent1")
    Order.write.mode("overwrite").saveAsTable("portrait.Order1")
    UserMsg.write.mode("overwrite").saveAsTable("portrait.UserMsg1")
    UserEvent.write.mode("overwrite").saveAsTable("portrait.UserEvent1")

  }
  def filterOName(frame:DataFrame): DataFrame ={
    return frame.filter(col("owner_name")=!="EA级" &&
                  col("owner_name")=!="EB级" &&
                  col("owner_name")=!="EC级" &&
                  col("owner_name")=!="ED级" &&
                  col("owner_name")=!="EE级" )
  }
  def filterOCode(frame:DataFrame): DataFrame ={
    return frame.where(col("owner_code") =!= "02")
      .where(col("owner_code") =!= "09")
      .where(col("owner_code") =!= "10").toDF()
  }
}
